import datetime
import logging
import time
import uuid

from kubernetes import watch
from myapp.app import app
from myapp.const.service import (
    DEPLOYTIMEOUT_WHEN_LACK_OF_RESOURCE,
    REDIS_KEY_DEPLOYTIMEOUT_WHEN_LACK_OF_RESOURCE,
    ServiceStatus,
)
from myapp.models.model_serving import InferenceService, Service
from myapp.third.k8s.py_karmada import get_k8s_client_by_region
from myapp.third.k8s.tools import ignore_404
from myapp.third.redis.client import RedisClient
from myapp.utils.exception import ignore_exception, log_exception
from myapp.utils.log import set_task_id
from myapp.utils.sess import session_scope


conf = app.config

log = logging.getLogger(__name__)


def listen_service(region='default'):
    logging.warning(f'[listen_service] start. region: {region} ')

    namespace = conf.get('SERVICE_NAMESPACE')
    client = get_k8s_client_by_region(region=region, action='search')

    log.info('begin listen service')
    w = watch.Watch()
    while True:
        try:
            for event in w.stream(
                client.AppsV1Api.list_deployment_for_all_namespaces,
                timeout_seconds=60,
            ):
                request_id = str(uuid.uuid4())
                set_task_id(request_id)
                deployment = event['object']
                ins_name = deployment.metadata.name

                ns = deployment.metadata.namespace
                if ns.startswith(namespace) is False:
                    continue

                sys_err_msg = f'系统异常，请联系技术支持或提交反馈信息（{request_id}）'
                with session_scope() as session:
                    ins = (
                        session.query(InferenceService)
                        .filter(
                            InferenceService.name == ins_name,
                            InferenceService.status.in_(
                                [
                                    ServiceStatus.deploying.value,
                                    ServiceStatus.upgrading.value,
                                    ServiceStatus.running.value,
                                ]
                            ),
                        )
                        .first()
                    )
                    if ins is None:
                        continue

                    deployment = None
                    with ignore_404:
                        deployment = client.AppsV1Api.read_namespaced_deployment_status(
                            name=ins_name, namespace=ns
                        )

                    if deployment is None:
                        log.error(f'获取deployment失败，{ins_name}')
                        update_service_status(
                            ins.id,
                            ins.service_id,
                            ins.status,
                            ServiceStatus.failed.value,
                            sys_err_msg,
                            session,
                        )
                        continue

                    replicas = deployment.spec.replicas
                    ready_replicas = 0
                    with ignore_exception:
                        ready_replicas = deployment.status.ready_replicas
                    # 升级过程中，ready_replicas不能反映真实的就绪pods数量
                    real_ready_replicas = calc_real_ready_replicas(deployment)

                    if real_ready_replicas == deployment.status.replicas:
                        log.info(f'Deployment全部pod已就绪,{ins.name,}')

                        update_service_status(
                            ins.id,
                            ins.service_id,
                            ins.status,
                            ServiceStatus.running.value,
                            '',
                            session,
                        )
                        continue

                    if ready_replicas == 0 and ins.status in [
                        ServiceStatus.running.value,
                        ServiceStatus.upgrading.value,
                    ]:
                        log.info(f'Deployment无就绪实例,{ins.name}')

                        update_service_status(
                            ins.id,
                            ins.service_id,
                            ins.status,
                            ins.status,
                            '未知异常，异常实例自动重启中。',
                            session,
                        )
                        continue

                    if ins.status in [ServiceStatus.running.value]:
                        if real_ready_replicas < replicas:
                            log.info(f'Deployment就绪实例数小于期望实例数,{ins.name}')

                            update_service_status(
                                ins.id,
                                ins.service_id,
                                ins.status,
                                ins.status,
                                '未知异常，异常实例自动重启中。',
                                session,
                            )
                            continue

                    error_msg = ''
                    with log_exception:
                        error_msg = get_pods_error_msg(ins, client, ns, sys_err_msg)
                    if real_ready_replicas >= 1:
                        log.info(f'存在已就绪pod,{ins.name,}')

                        update_service_status(
                            ins.id,
                            ins.service_id,
                            ins.status,
                            ServiceStatus.running.value,
                            error_msg,
                            session,
                        )
                        continue

                    if len(error_msg) > 0 and ins.status == ServiceStatus.deploying.value:
                        log.info(f'{ins.service_id}服务部署失败，error_msg:{error_msg}')

                        update_service_status(
                            ins.id,
                            ins.service_id,
                            ins.status,
                            ServiceStatus.failed.value,
                            error_msg,
                            session,
                        )
                        continue

                    # 检测部署超时
                    if ins.status == ServiceStatus.deploying.value:
                        deploy_start_time = ins.service.deploy_start_time
                        if deploy_start_time:
                            # 计算部署时间(分钟),30分钟就算部署超时
                            time_diff = datetime.datetime.now() - deploy_start_time
                            if time_diff.total_seconds() > ins.service.deploy_timeout:
                                log.info(
                                    '部署超时,请联系系统管理员。'
                                    f'开始部署时刻：{deploy_start_time},'
                                    f' now：{datetime.datetime.now()}'
                                )

                                update_service_status(
                                    ins.id,
                                    ins.service_id,
                                    ins.status,
                                    ServiceStatus.failed.value,
                                    '部署超时，请稍后重试。',
                                    session,
                                )
                                continue
        except Exception:
            log.error('listen_service error', exc_info=True)
            w = watch.Watch()
            time.sleep(60)


def update_service_status(ins_id, srv_id, old_status, new_status, err_msg, session):
    fields = {'status': new_status, 'error_msg': err_msg}
    if old_status == ServiceStatus.deploying.value and new_status == ServiceStatus.running.value:
        logging.info('每次服务停止或者删除的时候，触发计费')
        fields['running_start_time'] = datetime.datetime.now()

    res = (
        session.query(Service)
        .filter(Service.id == srv_id, Service.status == old_status)
        .update(fields, synchronize_session=False)
    )
    if res > 0:
        session.query(InferenceService).filter(InferenceService.id == ins_id).update(
            {'status': new_status, 'error_msg': err_msg}
        )


def calc_real_ready_replicas(deployment):
    ready_replicas = 0
    with ignore_exception:
        ready_replicas = deployment.status.ready_replicas
        if ready_replicas is None:
            ready_replicas = 0

    unavailable_replicas = 0
    with ignore_exception:
        unavailable_replicas = deployment.status.unavailable_replicas
        if unavailable_replicas is None:
            unavailable_replicas = 0

    real_ready_replicas = ready_replicas - unavailable_replicas
    if real_ready_replicas < 0:
        real_ready_replicas = 0
    return real_ready_replicas


def get_pods_error_msg(ins, client, ns, sys_err_msg):
    pods = None
    with log_exception:
        pods = client.v1.list_namespaced_pod(namespace=ns, label_selector=f'app={ins.name}')

    if pods is None or len(pods.items) == 0:
        log.error('找不到对应的 k8s pod')
        return '找不到对应的 k8s pods'

    for pod in pods.items:
        conditions = {}
        for cond in pod.status.conditions:
            conditions[cond.type] = cond

        # https://blog.51cto.com/u_15127705/4273386
        # PodScheduled 指示pod是否已被调度到某个节点。
        cond = conditions.get('PodScheduled')
        if cond.status == 'False':
            log.info(
                'PodScheduled'
                f' Failed,type:{cond.type},reason:{cond.reason},'
                f' message:{cond.message}'
            )

            # 资源不足
            if pod.status.phase == 'Pending' and cond.reason == 'Unschedulable':
                log.info(f'资源不足，{cond}')
                deploy_start_time = ins.service.deploy_start_time
                if deploy_start_time is None:
                    return '推理资源不足，请稍后重试。'

                timeout = RedisClient.get(REDIS_KEY_DEPLOYTIMEOUT_WHEN_LACK_OF_RESOURCE)
                timeout = DEPLOYTIMEOUT_WHEN_LACK_OF_RESOURCE if timeout is None else timeout
                time_diff = datetime.datetime.now() - deploy_start_time
                if time_diff.total_seconds() > int(timeout):
                    return '推理资源不足，请稍后重试。'

            continue

        # Initialized 指示pod是否已被调度到某个节点。
        cond = conditions.get('Initialized')
        if cond.status == 'False':
            log.info(
                'Initialized'
                f' Failed,type:{cond.type},reason:{cond.reason},'
                f' message:{cond.message}'
            )

            continue

        # ContainersReady  Pod里的所有容器都表明已经准备好了。
        # 这是整个Pod准备就绪的必要条件，但不是充分条件。
        cond = conditions.get('ContainersReady')
        if cond.status == 'False':
            for container_status in pod.status.container_statuses:
                if container_status.ready:
                    continue
                log.info(container_status)

                if container_status.state.waiting:
                    state = container_status.state.waiting

                    if state.reason == 'ImagePullBackOff' and pod.status.phase == 'Pending':
                        log.info(
                            f'镜像拉取失败{container_status.image},reason:' ' ImagePullBackOff'
                        )
                        return '镜像地址异常或镜像已失效。'

                    if (
                        state.reason == 'CrashLoopBackOff'
                        and pod.status.phase == 'Running'
                        and container_status.restart_count >= 5
                    ):
                        log.info(
                            '服务多次异常退出，'
                            f'重启次数:{container_status.restart_count},reason:'
                            ' CrashLoopBackOff'
                        )
                        return '模型内部故障，请修复模型后重试。'

                    if state.reason == 'InvalidImageName' and pod.status.phase == 'Pending':
                        log.info(
                            '无效的镜像地址，'
                            f'镜像地址：{container_status.image},reason:'
                            ' InvalidImageName'
                        )
                        return '镜像地址异常或镜像已失效。'

                elif container_status.state.running:
                    pass
                elif container_status.state.terminated:
                    pass
                else:
                    log.error(f'unknown state: {container_status.state}')

            continue

        # Ready pod已经准备好为其客户端提供服务。
        # Pod内的容器和Pod内的readiness gates都已就绪。
        cond = conditions.get('Ready')
        if cond.status == 'False':
            log.error(f'[{cond.type}]:{cond.reason}, {cond.message}')
            return sys_err_msg


# 不能使用异步io，因为stream会阻塞
if __name__ == '__main__':
    listen_service()
